package com.kmxd.ams.app.arrange.command;

import cn.easyes.core.conditions.select.LambdaEsQueryWrapper;
import cn.easyes.core.toolkit.FieldUtils;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.io.unit.DataSizeUtil;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.kmxd.ams.app.event.es.command.EsOcrCmd;
import com.kmxd.ams.client.archive.IArchiveAuthSvc;
import com.kmxd.ams.client.archive.IArchiveTypeSvc;
import com.kmxd.ams.client.archive.dto.ArchiveTypeQueryDTO;
import com.kmxd.ams.client.archive.vo.ArchiveAuthScopeVO;
import com.kmxd.ams.client.archive.vo.ArchiveTypeListVO;
import com.kmxd.ams.client.archive.vo.ArchiveTypeVO;
import com.kmxd.ams.client.arrange.IBoxSvc;
import com.kmxd.ams.client.arrange.IDocumentSvc;
import com.kmxd.ams.client.arrange.IProjectSvc;
import com.kmxd.ams.client.arrange.IVolumeSvc;
import com.kmxd.ams.client.arrange.dto.CreateEsIndexDTO;
import com.kmxd.ams.client.es.*;
import com.kmxd.ams.client.event.es.EsCreateIndexEvent;
import com.kmxd.ams.client.event.es.RemoveEsDataEvent;
import com.kmxd.ams.client.oss.IEfileRecordSvc;
import com.kmxd.ams.client.oss.vo.EfileRecordListVO;
import com.kmxd.ams.client.system.ISysCropSvc;
import com.kmxd.ams.client.system.ISysFondsSvc;
import com.kmxd.ams.client.system.vo.SysCropVO;
import com.kmxd.ams.client.system.vo.SysFondsVO;
import com.kmxd.ams.core.common.constant.*;
import com.kmxd.ams.core.common.enums.ArchModelEnum;
import com.kmxd.ams.core.common.enums.ArchStatusEnum;
import com.kmxd.ams.core.common.enums.CreateEsIndexEnum;
import com.kmxd.ams.core.exception.BizTipException;
import com.kmxd.ams.core.util.ConvertUtil;
import com.kmxd.ams.core.util.DynamicTableNameUtil;
import com.kmxd.ams.core.util.IdUtil;
import com.kmxd.ams.infra.arrange.base.BaseBO;
import com.kmxd.ams.infra.arrange.base.SystemBO;
import com.kmxd.ams.infra.arrange.entity.Box;
import com.kmxd.ams.infra.arrange.entity.Document;
import com.kmxd.ams.infra.arrange.entity.Project;
import com.kmxd.ams.infra.arrange.entity.Volume;
import com.kmxd.ams.infra.search.entity.*;
import com.kmxd.ams.infra.tmpl.dto.ArchTmplBO;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import javax.validation.constraints.NotNull;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * es操作
 *
 * @author nickbi
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class EsCmd {

  private final IArchiveTypeSvc archiveTypeSvc;
  private final IDocumentEsSvc documentEsSvc;
  private final IVolumeEsSvc volumeEsSvc;
  private final IBoxEsSvc boxEsSvc;
  private final IProjectEsSvc projectEsSvc;
  private final IVolumeSvc volumeSvc;
  private final IDocumentSvc documentSvc;
  private final IBoxSvc boxSvc;
  private final IProjectSvc projectSvc;
  private final IArchiveAuthSvc archiveAuthSvc;
  private final IEfileRecordSvc efileRecordSvc;
  private final EsOcrCmd esOcrCmd;
  private final ISysFondsSvc sysFondsSvc;

  /** 副本 */
  @Value("${easy-es.global-config.replicas:2}")
  private Integer replicas;

  /** 分片 */
  @Value("${easy-es.global-config.shards:3}")
  private Integer shards;

  /** 批次数量 */
  @Value("${easy-es.global-config.batchSize:10000}")
  private Integer esBatchSize;

  @Autowired
  @Qualifier("esForkJoinPool")
  private ForkJoinPool esForkJoinPool;

  public static final String START_INDEX_TMP = "开始创建{}索引,总数据量：{},批操作数量：{},共:{}批次";
  public static final String BATCH_LOG_TMP = "开始创建{}索引,批次：{}";
  public static final String BATCH_LOG_TMP_RES = "开始创建{}索引,批次：{},结果：{}";

  @Transactional(rollbackFor = Exception.class)
  public void createIndex(@NotNull CreateEsIndexDTO req) {
    if (CreateEsIndexEnum.ARCHIVE_TYPE.equals(req.getType())) {
      ArchiveTypeVO archiveType = archiveTypeSvc.view(req.getArchTypeId());
      if (Objects.isNull(archiveType)) {
        throw BizTipException.instance(ErrorCode.NOT_FOUND, "未找到门类信息");
      }
      createIndex(archiveType);
    } else {
      ArchiveTypeQueryDTO qryArchiveType = new ArchiveTypeQueryDTO();
      qryArchiveType.setFondsId(req.getFondsId());
      qryArchiveType.setComId(req.getComId());
      List<ArchiveTypeListVO> archiveTypeList = archiveTypeSvc.selectList(qryArchiveType);
      archiveTypeList.forEach(
          archiveType -> {
            ArchiveTypeVO type = ConvertUtil.toBean(archiveType, ArchiveTypeVO.class);
            createIndex(type);
          });
    }
  }

  private void createIndex(ArchiveTypeVO archiveType) {
    if (Boolean.TRUE.equals(archiveType.getHaveVol())) {
      CompletableFuture.runAsync(() -> createVolIndex(archiveType), esForkJoinPool);
    }
    if (Boolean.TRUE.equals(archiveType.getHaveBox())) {
      CompletableFuture.runAsync(() -> createBoxIndex(archiveType), esForkJoinPool);
    }
    if (Boolean.TRUE.equals(archiveType.getHaveProject())) {
      CompletableFuture.runAsync(() -> createProjectIndex(archiveType), esForkJoinPool);
    }
    CompletableFuture.runAsync(() -> createDocIndex(archiveType), esForkJoinPool);
  }

  private void createVolIndex(ArchiveTypeVO archiveType) {
    BaseBO qryVol = new BaseBO();
    qryVol.setArchTypeId(archiveType.getId());
    qryVol.setTableId(archiveType.getVolTableId());
    if (volumeEsSvc.existsIndex(qryVol)) {
      volumeEsSvc.removeIndex(qryVol);
    }
    boolean index = volumeEsSvc.createIndex(qryVol, true, shards, replicas);
    if (!index) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      volumeEsSvc.createIndex(qryVol, true, shards, replicas);
    }
    Volume qryVolPage = new Volume();
    qryVolPage.setTableId(archiveType.getVolTableId());
    qryVolPage.setArchTypeId(archiveType.getId());
    qryVolPage.setStatus(ArchStatusEnum.S1.getCode());
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(archiveType.getId());
    DynamicTableNameUtil.setTableName(archiveType.getVolTableId());
    long count = volumeSvc.count(qryVolPage);
    if (ObjectUtil.equals(count, 0L)) {
      return;
    }
    long loopNumber = getLoopNumber(count);
    log.info(START_INDEX_TMP, archiveType.getVolTableId(), count, esBatchSize, loopNumber);
    for (int i = 0; i < loopNumber; i++) {
      int finalI = i;
      CompletableFuture.runAsync(
          () -> {
            log.info("开始创建{},批次：{}", archiveType.getVolTableId(), finalI + 1);
            Page<Volume> page = new Page<>(finalI + 1, esBatchSize);
            DynamicTableNameUtil.setTableName(archiveType.getVolTableId());
            Page<Volume> resPage = volumeSvc.selectPage(page, qryVolPage);
            List<Volume> list = resPage.getRecords();
            if (ObjectUtil.isEmpty(resPage) || ObjectUtil.isEmpty(list)) {
              return;
            }
            save(list, sysFonds, archiveType.getVolTableId(), archiveType);
          },
          esForkJoinPool);
    }
  }

  private void createDocIndex(ArchiveTypeVO archiveType) {
    BaseBO qryDoc = new BaseBO();
    qryDoc.setArchTypeId(archiveType.getId());
    qryDoc.setTableId(archiveType.getDocTableId());
    if (documentEsSvc.existsIndex(qryDoc)) {
      documentEsSvc.removeIndex(qryDoc);
    }
    boolean index = documentEsSvc.createIndex(qryDoc, true, shards, replicas);
    if (!index) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      documentEsSvc.createIndex(qryDoc, true, shards, replicas);
    }
    Document qryDocPage = new Document();
    qryDocPage.setTableId(archiveType.getDocTableId());
    qryDocPage.setArchTypeId(archiveType.getId());
    qryDocPage.setStatus(ArchStatusEnum.S1.getCode());
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(archiveType.getId());
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    DynamicTableNameUtil.setTableName(archiveType.getDocTableId());
    long count = documentSvc.count(qryDocPage);
    if (ObjectUtil.equals(count, 0L)) {
      return;
    }

    long loopNumber = getLoopNumber(count);
    log.info(START_INDEX_TMP, archiveType.getDocTableId(), count, esBatchSize, loopNumber);
    for (int i = 0; i < loopNumber; i++) {
      int finalI = i;
      CompletableFuture.runAsync(
          () -> {
            log.info("开始创建{},批次：{}", archiveType.getDocTableId(), finalI + 1);
            Page<Document> page = new Page<>(finalI + 1, esBatchSize);
            DynamicTableNameUtil.setTableName(archiveType.getDocTableId());
            Page<Document> resPage = documentSvc.selectPage(page, qryDocPage);
            List<Document> list = resPage.getRecords();
            if (ObjectUtil.isEmpty(resPage) || ObjectUtil.isEmpty(list)) {
              return;
            }
            save(list, sysFonds, archiveType.getDocTableId(), archiveType);
          },
          esForkJoinPool);
    }
  }

  /**
   * 获取循环次数
   *
   * @param count
   * @return
   */
  private long getLoopNumber(long count) {
    long pageIndex;
    boolean noMore = count % esBatchSize == 0;
    if (noMore) {
      pageIndex = count / esBatchSize;
    } else {
      pageIndex = count / esBatchSize + 1;
    }
    return pageIndex;
  }

  /**
   * 获取电子文件信息
   *
   * @param sysFonds
   * @param fileList
   * @return
   */
  private List<EFileRecordEs> getEFileInfo(SysFondsVO sysFonds, List<EfileRecordListVO> fileList) {
    try {
      String tableName = Constants.TABLE_NAME_EFILE + "_" + sysFonds.getAutoIndex();
      DynamicTableNameUtil.setTableName(tableName);
      if (ObjectUtil.isEmpty(fileList)) {
        return Lists.newArrayList();
      }
      return fileList.stream()
          .map(
              file -> {
                EFileRecordEs vo = new EFileRecordEs();
                String content = esOcrCmd.getOcrContentInfo(file);
                vo.setOcrContent(content);
                vo.setOcrContentKw(CharSequenceUtil.sub(content, 0, 3000));
                vo.setTitle(file.getTitle());
                vo.setTitleKw(file.getTitle());
                vo.setSize(DataSizeUtil.format(NumberUtil.parseLong(file.getFileSize())));
                vo.setExt(file.getExt());
                vo.setId(file.getId());
                return vo;
              })
          .collect(Collectors.toList());
    } catch (Exception e) {
      log.error(e.getMessage(), e);
      return Lists.newArrayList();
    }
  }

  private void createBoxIndex(ArchiveTypeVO archiveType) {
    BaseBO qryDoc = new BaseBO();
    qryDoc.setArchTypeId(archiveType.getId());
    qryDoc.setTableId(archiveType.getBoxTableId());
    if (documentEsSvc.existsIndex(qryDoc)) {
      documentEsSvc.removeIndex(qryDoc);
    }
    boolean index = boxEsSvc.createIndex(qryDoc, true, shards, replicas);
    if (!index) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      boxEsSvc.createIndex(qryDoc, true, shards, replicas);
    }
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(archiveType.getId());
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    Box qryBoxPage = new Box();
    qryBoxPage.setTableId(archiveType.getBoxTableId());
    qryBoxPage.setArchTypeId(archiveType.getId());
    qryBoxPage.setStatus(ArchStatusEnum.S1.getCode());
    DynamicTableNameUtil.setTableName(archiveType.getBoxTableId());
    long count = boxSvc.count(qryBoxPage);
    if (ObjectUtil.equals(count, 0L)) {
      return;
    }
    long loopNumber = getLoopNumber(count);
    log.info(START_INDEX_TMP, archiveType.getBoxTableId(), count, esBatchSize, loopNumber);

    for (int i = 0; i < loopNumber; i++) {
      int finalI = i;
      CompletableFuture.runAsync(
          () -> {
            log.info("开始创建{},批次：{}", archiveType.getDocTableId(), finalI + 1);
            Page<Document> page = new Page<>(finalI + 1, esBatchSize);
            DynamicTableNameUtil.setTableName(archiveType.getBoxTableId());
            Page<Box> resPage = boxSvc.selectPage(page, qryBoxPage);
            List<Box> list = resPage.getRecords();
            if (ObjectUtil.isEmpty(resPage) || ObjectUtil.isEmpty(list)) {
              return;
            }
            save(list, sysFonds, archiveType.getBoxTableId(), archiveType);
          },
          esForkJoinPool);
    }
  }

  private void createProjectIndex(ArchiveTypeVO archiveType) {
    BaseBO qryProj = new BaseBO();
    qryProj.setArchTypeId(archiveType.getId());
    qryProj.setTableId(archiveType.getProjTableId());
    if (projectEsSvc.existsIndex(qryProj)) {
      projectEsSvc.removeIndex(qryProj);
    }
    boolean index = projectEsSvc.createIndex(qryProj, true, shards, replicas);
    if (!index) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      projectEsSvc.createIndex(qryProj, true, shards, replicas);
    }
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(archiveType.getId());
    // 项目不区分状态，全部索引
    ImmutableSet.of(ArchStatusEnum.S0.getCode(), ArchStatusEnum.S1.getCode())
        .forEach(
            status -> {
              Project qryProjPage = new Project();
              qryProjPage.setTableId(archiveType.getProjTableId());
              qryProjPage.setArchTypeId(archiveType.getId());
              qryProjPage.setStatus(status);
              DynamicTableNameUtil.setTableName(archiveType.getProjTableId());
              long count = projectSvc.count(qryProjPage);
              if (ObjectUtil.equals(count, 0L)) {
                return;
              }
              ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
              SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
              long loopNumber = getLoopNumber(count);
              log.info(
                  START_INDEX_TMP, archiveType.getProjTableId(), count, esBatchSize, loopNumber);
              for (int i = 0; i < loopNumber; i++) {
                int finalI = i;
                CompletableFuture.runAsync(
                    () -> {
                      log.info("开始创建{},批次：{}", archiveType.getDocTableId(), finalI + 1);
                      Page<Project> page = new Page<>(finalI + 1, esBatchSize);
                      DynamicTableNameUtil.setTableName(archiveType.getProjTableId());
                      Page<Project> resPage = projectSvc.selectPage(page, qryProjPage);
                      List<Project> list = resPage.getRecords();
                      if (ObjectUtil.isEmpty(resPage) || ObjectUtil.isEmpty(list)) {
                        return;
                      }
                      save(list, sysFonds, archiveType.getProjTableId(), archiveType);
                    });
              }
            });
  }

  public <T extends SystemBO> void save(Collection<Long> ids, Long archTypeId, String tableId) {
    ArchiveTypeVO archiveType = archiveTypeSvc.view(archTypeId);
    if (Objects.isNull(archiveType)) {
      return;
    }
    if (ObjectUtil.isEmpty(ids)) {
      return;
    }
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    ServiceInfo serviceInfo = getServiceInfo(tableId);
    List<Long> list = serviceInfo.serviceEs.checkDataExist(ids, tableId);
    if (ObjectUtil.isNotEmpty(list)) {
      ids.removeAll(list);
    }
    if (ObjectUtil.isNotEmpty(list)) {
      List<SystemBO> archList = getArchByIds(list, tableId);
      update(archList, tableId, sysFonds, archiveType);
    }
    if (ObjectUtil.isEmpty(ids)) {
      return;
    }

    List<SystemBO> data = getArchByIds(ids, tableId);
    save(data, sysFonds, tableId, archiveType);
  }

  public <T extends SystemBO> void update(Collection<Long> ids, Long archTypeId, String tableId) {
    ArchiveTypeVO archiveType = archiveTypeSvc.view(archTypeId);
    if (Objects.isNull(archiveType)) {
      return;
    }
    if (ObjectUtil.isEmpty(ids)) {
      return;
    }
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    List<T> data = getArchByIds(ids, tableId);
    update(data, tableId, sysFonds, archiveType);
  }

  public <T extends SystemBO> void remove(List<T> list, Long archTypeId, String tableId) {
    ArchiveTypeVO archiveType = archiveTypeSvc.view(archTypeId);
    if (Objects.isNull(archiveType)) {
      return;
    }
    if (ObjectUtil.isEmpty(list)) {
      return;
    }
    ISysFondsSvc sysFondsSvc = SpringUtil.getBean(ISysFondsSvc.class);
    SysFondsVO sysFonds = sysFondsSvc.view(archiveType.getFondsId());
    update(list, tableId, sysFonds, archiveType);
  }

  private EsCmd.ServiceInfo getServiceInfo(String tableId) {

    if (tableId.contains(ArchTmplBO.DOCUMENT)) {
      return new EsCmd.ServiceInfo(documentEsSvc, DocumentEs.class);
    }
    if (tableId.contains(ArchTmplBO.VOLUME)) {
      return new EsCmd.ServiceInfo(volumeEsSvc, VolumeEs.class);
    }
    if (tableId.contains(ArchTmplBO.BOX)) {

      return new EsCmd.ServiceInfo(boxEsSvc, BoxEs.class);
    }
    if (tableId.contains(ArchTmplBO.PROJECT)) {
      return new EsCmd.ServiceInfo(projectEsSvc, ProjectEs.class);
    }
    return new EsCmd.ServiceInfo();
  }

  /**
   * 创建索引
   *
   * @param list 数据列表
   * @param tableId 表名
   * @param archiveType 门类
   * @param <T>
   * @param <R>
   */
  private <T extends SystemBO, R> void save(
      List<T> list, SysFondsVO sysFonds, String tableId, ArchiveTypeVO archiveType) {
    BaseBO qryDoc = new BaseBO();
    qryDoc.setArchTypeId(archiveType.getId());
    qryDoc.setTableId(tableId);
    // 检查index是否已经存在,存在不创建,不存在创建index
    ServiceInfo serviceInfo = getServiceInfo(tableId);
    IServiceEs serviceEs = serviceInfo.getServiceEs();
    boolean indexFlag = serviceEs.createIndex(qryDoc, false, shards, replicas);
    log.info("创建es索引结果:{},参数:{}", indexFlag, qryDoc);
    if (!indexFlag) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      boolean index = serviceEs.createIndex(qryDoc, false, shards, replicas);
      log.info("重建es索引结果:{},参数:{}", index, qryDoc);
    }
    ISysCropSvc sysCropSvc = SpringUtil.getBean(ISysCropSvc.class);
    SysCropVO sysCropVO = sysCropSvc.view(sysFonds.getComId());
    Map<Long, List<EfileRecordListVO>> recordMap = Maps.newHashMap();
    if (CharSequenceUtil.contains(tableId, ArchTmplBO.DOCUMENT)) {
      Set<Long> docIds = list.stream().map(SystemBO::getId).collect(Collectors.toSet());
      String tableName = Constants.TABLE_NAME_EFILE + "_" + sysFonds.getAutoIndex();
      DynamicTableNameUtil.setTableName(tableName);
      List<EfileRecordListVO> fileList =
          Optional.ofNullable(efileRecordSvc.selectListByDocIds(archiveType.getId(), docIds))
              .orElse(Lists.newArrayList());
      recordMap =
          fileList.stream()
              .filter(item -> ObjectUtil.equals(archiveType.getId(), item.getArchTypeId()))
              .collect(Collectors.groupingBy(EfileRecordListVO::getArchId));
    }

    Map<Long, List<EfileRecordListVO>> finalRecordMap = recordMap;
    List<R> dataList =
        list.stream()
            .filter(
                data ->
                    // 除项目级外，其他档案校验归档状态
                    CharSequenceUtil.contains(tableId, ArchTmplBO.PROJECT)
                        || (!CharSequenceUtil.contains(tableId, ArchTmplBO.PROJECT)
                            && ObjectUtil.equals(data.getStatus(), ArchStatusEnum.S1.getCode())))
            .map(
                record -> {
                  record.setTableId(tableId);
                  record.setArchTypeId(archiveType.getId());
                  JSONObject parseObj = JSONUtil.parseObj(record);
                  parseObj.set("archTypeId", archiveType.getId());
                  parseObj.set(ArchConst.TABLE_ID, tableId);
                  parseObj.set(ArchConst.COM_ID, archiveType.getComId());
                  parseObj.set(ArchConst.COM_NAME, sysCropVO.getName());
                  parseObj.set("archTypeName", archiveType.getName());
                  //                  parseObj.set(ArchConst.DEF_FONDS_ID_FIELD,
                  // archiveType.getFondsId());
                  //                  parseObj.set(ArchConst.OSS_SETTING_FONDS_KEY,
                  // sysFonds.getName());
                  parseObj.set("archModel", archiveType.getModel());
                  parseObj.set(
                      "archModelName", ArchModelEnum.convert(archiveType.getModel()).getDesc());
                  if (tableId.contains(ArchTmplBO.DOCUMENT)) {
                    List<EFileRecordEs> esContent =
                        getEFileInfo(sysFonds, finalRecordMap.get(record.getId()));
                    parseObj.set(ArchConst.ES_AMS_RECORDS, esContent);

                    parseObj.set(
                        FieldUtils.val(DocumentEs::getTitleKw),
                        parseObj.getStr(ArchConst.FIELD_TITLE));
                    parseObj.set(
                        FieldUtils.val(DocumentEs::getArchCodeKw),
                        parseObj.getStr(ArchConst.ARCH_CODE));
                    parseObj.set(
                        FieldUtils.val(Document::getDocSequence) + EsConst.FIELD_NUMBER_SUFFIX,
                        parseObj.getStr(ArchConst.DOC_SEQUENCE));
                  }

                  // 案卷
                  if (tableId.contains(ArchTmplBO.VOLUME)) {
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getTitleKw),
                        parseObj.getStr(ArchConst.FIELD_TITLE));
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getArchCodeKw),
                        parseObj.getStr(ArchConst.ARCH_CODE));
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getVolSequence) + EsConst.FIELD_NUMBER_SUFFIX,
                        parseObj.getStr(ArchConst.VOL_SEQUENCE));
                  }

                  buildAuthInfo(record, parseObj);
                  Class<R> clazz = serviceInfo.getClazz();
                  return parseObj.toBean(clazz);
                })
            .collect(Collectors.toList());
    boolean flag = serviceEs.saveBatch(dataList, tableId);
    AtomicInteger index = new AtomicInteger(1);
    while (!flag) {
      ThreadUtil.sleep(2000);
      String ids =
          list.stream().map(bo -> String.valueOf(bo.getId())).collect(Collectors.joining(","));
      log.debug("门类：{},表：{},数据:{},索引失败，开始重建索引...", archiveType.getId(), tableId, ids);
      int andIncrement = index.getAndIncrement();

      if (andIncrement == 3) {
        return;
      }
      boolean batch = serviceEs.saveBatch(dataList, tableId);
      log.debug("门类：{},表：{},数据:{},重建索引结果：{}", archiveType.getId(), tableId, ids, batch);
      if (batch) {
        return;
      }
    }
  }

  private <T extends SystemBO> void buildAuthInfo(T record, JSONObject parseObj) {
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(record.getArchTypeId());
    if (Objects.isNull(auth)) {
      return;
    }
    // todo... 知悉范围重构
    Map<String, Object> map = getAuthInfo(auth);
    map.forEach(parseObj::set);
  }

  private Map<String, Object> getAuthInfo(ArchiveAuthScopeVO auth) {
    Map<String, Object> map = BeanUtil.beanToMap(auth);
    map.remove(ArchConst.DEF_ID);
    map.remove(ArchConst.COM_ID);
    map.remove(ArchConst.DEF_FONDS_ID_FIELD);
    map.remove(ArchConst.ARCH_TYPE_ID);
    map.remove(FieldUtils.val(ArchiveAuthScopeVO::getScopeType));
    return map;
  }

  /**
   * 更新es数据
   *
   * @param list
   * @param tableId
   * @param archiveType
   * @param <T>
   * @param <R>
   */
  private <T extends SystemBO, R> void update(
      List<T> list, String tableId, SysFondsVO sysFonds, ArchiveTypeVO archiveType) {
    if (ObjectUtil.isEmpty(list)) {
      log.info("es数据更新数据为空");
      return;
    }
    ServiceInfo serviceInfo = getServiceInfo(tableId);
    Map<Long, List<EfileRecordListVO>> recordMap = Maps.newHashMap();
    if (CharSequenceUtil.contains(tableId, ArchTmplBO.DOCUMENT)) {
      Set<Long> docIds = list.stream().map(SystemBO::getId).collect(Collectors.toSet());
      String tableName = Constants.TABLE_NAME_EFILE + "_" + sysFonds.getAutoIndex();
      DynamicTableNameUtil.setTableName(tableName);
      List<EfileRecordListVO> fileList =
          Optional.ofNullable(efileRecordSvc.selectListByDocIds(archiveType.getId(), docIds))
              .orElse(Lists.newArrayList());

      recordMap =
          fileList.stream()
              .filter(item -> ObjectUtil.equals(archiveType.getId(), item.getArchTypeId()))
              .collect(Collectors.groupingBy(EfileRecordListVO::getArchId));
    }
    ISysCropSvc sysCropSvc = SpringUtil.getBean(ISysCropSvc.class);
    SysCropVO sysCropVO = sysCropSvc.view(sysFonds.getComId());
    Map<Long, List<EfileRecordListVO>> finalRecordMap = recordMap;
    List<R> dataList =
        list.stream()
            .filter(
                data -> // 除项目级外，其他档案校验归档状态
                CharSequenceUtil.contains(tableId, ArchTmplBO.PROJECT)
                        || (!CharSequenceUtil.contains(tableId, ArchTmplBO.PROJECT)
                            && ObjectUtil.equals(data.getStatus(), ArchStatusEnum.S1.getCode())))
            .map(
                record -> {
                  record.setTableId(tableId);
                  record.setArchTypeId(archiveType.getId());
                  JSONObject parseObj = JSONUtil.parseObj(record);
                  parseObj.set(ArchConst.TABLE_ID, tableId);
                  parseObj.set("archTypeId", archiveType.getId());
                  parseObj.set("archTypeName", archiveType.getName());
                  parseObj.set("archModel", archiveType.getModel());
                  parseObj.set(ArchConst.COM_ID, archiveType.getComId());
                  parseObj.set(ArchConst.COM_NAME, sysCropVO.getName());
                  //                  parseObj.set(ArchConst.OSS_SETTING_FONDS_KEY,
                  // sysFonds.getName());
                  //                  parseObj.set(ArchConst.DEF_FONDS_ID_FIELD, sysFonds.getId());
                  parseObj.set(
                      "archModelName", ArchModelEnum.convert(archiveType.getModel()).getDesc());
                  buildAuthInfo(record, parseObj);
                  // ocr原文
                  if (tableId.contains(ArchTmplBO.DOCUMENT)) {
                    List<EFileRecordEs> esContent =
                        getEFileInfo(sysFonds, finalRecordMap.get(record.getId()));
                    parseObj.set(ArchConst.ES_AMS_RECORDS, esContent);
                    parseObj.set(
                        FieldUtils.val(DocumentEs::getTitleKw),
                        parseObj.getStr(ArchConst.FIELD_TITLE));
                    parseObj.set(
                        FieldUtils.val(DocumentEs::getArchCodeKw),
                        parseObj.getStr(ArchConst.ARCH_CODE));
                    parseObj.set(
                        FieldUtils.val(Document::getDocSequence) + EsConst.FIELD_NUMBER_SUFFIX,
                        parseObj.getStr(ArchConst.DOC_SEQUENCE));
                  }
                  // 案卷
                  if (tableId.contains(ArchTmplBO.VOLUME)) {
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getTitleKw),
                        parseObj.getStr(ArchConst.FIELD_TITLE));
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getArchCodeKw),
                        parseObj.getStr(ArchConst.ARCH_CODE));
                    parseObj.set(
                        FieldUtils.val(VolumeEs::getVolSequence) + EsConst.FIELD_NUMBER_SUFFIX,
                        parseObj.getStr(ArchConst.VOL_SEQUENCE));
                  }
                  Class<R> clazz = serviceInfo.getClazz();
                  return parseObj.toBean(clazz);
                })
            .collect(Collectors.toList());
    IServiceEs serviceEs = serviceInfo.getServiceEs();
    boolean flag = serviceEs.updateBatchById(dataList, tableId);
    AtomicInteger index = new AtomicInteger(1);
    while (!flag) {
      ThreadUtil.sleep(2000);
      log.debug("retry update index");
      int andIncrement = index.getAndIncrement();

      log.debug("retry update index number:{}", andIncrement);
      if (andIncrement == 3) {
        return;
      }

      boolean batch = serviceEs.updateBatchById(dataList, tableId);
      log.debug("retry update index result:{}", batch);
      if (batch) {
        return;
      }
    }
  }

  /**
   * 通过id删除数据
   *
   * @param ids 数据ids
   * @param tableId 表名
   */
  public void removeByIds(String ids, String tableId) {
    EsCmd.ServiceInfo serviceInfo = getServiceInfo(tableId);
    serviceInfo.serviceEs.removeByIds(IdUtil.split(ids), tableId);
  }

  /**
   * 更新es权限
   *
   * @param archTypeId
   * @param <T>
   */
  public <T extends SystemBO> void updateAuth(Long archTypeId) {
    ArchiveTypeVO archiveType = archiveTypeSvc.view(archTypeId);
    if (Objects.isNull(archiveType)) {
      return;
    }
    List<String> tableIds = org.apache.commons.compress.utils.Lists.newArrayList();
    if (StrUtil.isNotBlank(archiveType.getProjTableId())) {
      tableIds.add(archiveType.getProjTableId());
    }
    if (StrUtil.isNotBlank(archiveType.getVolTableId())) {
      tableIds.add(archiveType.getVolTableId());
    }
    if (StrUtil.isNotBlank(archiveType.getBoxTableId())) {
      tableIds.add(archiveType.getBoxTableId());
    }
    tableIds.add(archiveType.getDocTableId());
    ArchiveAuthScopeVO auth = archiveAuthSvc.getByArchTypeId(archTypeId);
    tableIds.forEach(
        tableId -> {
          EsCmd.ServiceInfo serviceInfo = getServiceInfo(tableId);
          LambdaEsQueryWrapper<T> wrapper = new LambdaEsQueryWrapper<T>();
          wrapper.index(tableId);
          wrapper.eq(ArchConst.ARCH_TYPE_ID, archiveType.getId());
          JSONObject parseObj = JSONUtil.createObj();
          // todo... 知悉范围重构
          Map<String, Object> map = getAuthInfo(auth);
          map.forEach(parseObj::set);
          Object bean = JSONUtil.toBean(parseObj, serviceInfo.clazz);
          serviceInfo.getServiceEs().updateByQuery(bean, wrapper, tableId);
        });
  }

  private <T extends SystemBO> List<T> getArchByIds(Collection ids, String tableId) {
    DynamicTableNameUtil.setTableName(tableId);
    if (ObjectUtil.contains(tableId, ArchTmplBO.DOCUMENT)) {
      return documentSvc.listByIds(ids);
    }
    if (ObjectUtil.contains(tableId, ArchTmplBO.VOLUME)) {
      return volumeSvc.listByIds(ids);
    }
    if (ObjectUtil.contains(tableId, ArchTmplBO.BOX)) {
      return boxSvc.listByIds(ids);
    }
    if (ObjectUtil.contains(tableId, ArchTmplBO.PROJECT)) {
      return projectSvc.listByIds(ids);
    }
    return org.apache.commons.compress.utils.Lists.newArrayList();
  }

  public void batchCreateIndex(EsCreateIndexEvent event) {
    String archTypeIds = event.getArchTypeIds();
    if (CharSequenceUtil.isBlank(archTypeIds)) {
      return;
    }
    Set<Long> archTypeIdList = IdUtil.split(archTypeIds);
    archTypeIdList.forEach(archTypeId -> createIndex(archTypeId, event.getCleanOldIndex()));
  }

  public void createIndex(Long archTypeId, boolean cleanOldIndex) {
    ArchiveTypeVO view = archiveTypeSvc.view(archTypeId);
    if (Objects.isNull(view)) {
      return;
    }
    log.info("开始创建门类es索引,门类id: {},门类名称: {}", view.getId(), view.getName());

    if (view.getHaveProject()) {
      ProjectEs doc = new ProjectEs();
      doc.setArchTypeId(view.getId());
      doc.setTableId(view.getProjTableId());
      boolean index = projectEsSvc.createIndex(doc, cleanOldIndex);
      if (!index) {
        ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
        projectEsSvc.createIndex(doc, cleanOldIndex);
      }
      log.info("创建项目es索引,项目表id: {} 结果: {}", view.getProjTableId(), index);
    }

    if (view.getHaveVol()) {
      VolumeEs doc = new VolumeEs();
      doc.setArchTypeId(view.getId());
      doc.setTableId(view.getVolTableId());
      boolean index = volumeEsSvc.createIndex(doc, cleanOldIndex);
      if (!index) {
        ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
        volumeEsSvc.createIndex(doc, cleanOldIndex);
      }
      log.info("创建案卷es索引,案卷表id: {} 结果: {}", view.getVolTableId(), index);
    }
    DocumentEs doc = new DocumentEs();
    doc.setArchTypeId(view.getId());
    doc.setTableId(view.getDocTableId());
    boolean index = documentEsSvc.createIndex(doc, cleanOldIndex);
    // 重试
    if (!index) {
      ThreadUtil.sleep(RedissonConst.WAIT_TIME, TimeUnit.SECONDS);
      documentEsSvc.createIndex(doc, cleanOldIndex);
    }
    log.info("创建文件es索引,文件表id: {} 结果: {}", view.getDocTableId(), index);
  }

  @Data
  @AllArgsConstructor
  @NoArgsConstructor
  class ServiceInfo {
    private IServiceEs serviceEs;
    private Class clazz;
  }

  public void removeData(RemoveEsDataEvent event) {
    if (CharSequenceUtil.isNotBlank(event.getDocIds())) {
      Set<Long> docIds = IdUtil.split(event.getDocIds());
      String docTableId = Optional.ofNullable(event.getDocTableId()).orElse(event.getTableId());
      documentEsSvc.removeByIds(docIds, docTableId);
    }
    if (CharSequenceUtil.isNotBlank(event.getVolIds())) {
      Set<Long> volIds = IdUtil.split(event.getVolIds());
      String tableId = Optional.ofNullable(event.getVolTableId()).orElse(event.getTableId());
      volumeEsSvc.removeByIds(volIds, tableId);
    }

    if (CharSequenceUtil.isNotBlank(event.getBoxIds())) {
      Set<Long> ids = IdUtil.split(event.getBoxIds());
      String tableId = Optional.ofNullable(event.getBoxTableId()).orElse(event.getTableId());
      boxEsSvc.removeByIds(ids, tableId);
    }

    if (CharSequenceUtil.isNotBlank(event.getProjIds())) {
      Set<Long> ids = IdUtil.split(event.getProjIds());
      String tableId = Optional.ofNullable(event.getProjTableId()).orElse(event.getTableId());
      projectEsSvc.removeByIds(ids, tableId);
    }
    // 删除 项目内 案卷内 盒内文件
    removeInnerDocument(event);
  }

  public void removeInnerDocument(RemoveEsDataEvent event) {
    ArchiveTypeVO archiveTypeVO = archiveTypeSvc.view(event.getArchTypeId());
    if (ObjectUtil.isNull(archiveTypeVO)) {
      return;
    }
    if (ObjectUtil.isAllEmpty(event.getProjIds(), event.getVolIds(), event.getBoxIds())) {
      return;
    }
    LambdaEsQueryWrapper<DocumentEs> wrapper = new LambdaEsQueryWrapper<>();
    wrapper.index(archiveTypeVO.getDocTableId());
    if (CharSequenceUtil.isNotBlank(event.getVolIds())) {
      Set<Long> volIds = IdUtil.split(event.getVolIds());
      wrapper.in(DocumentEs::getVolId, volIds);
    }
    if (CharSequenceUtil.isNotBlank(event.getBoxIds())) {
      Set<Long> boxIds = IdUtil.split(event.getBoxIds());
      wrapper.in(DocumentEs::getBoxId, boxIds);
    }
    if (CharSequenceUtil.isNotBlank(event.getProjIds())) {
      Set<Long> projIds = IdUtil.split(event.getProjIds());
      wrapper.in(DocumentEs::getProjId, projIds);
      // 删除项目内案卷 盒
      if (BooleanUtil.isTrue(archiveTypeVO.getHaveVol())) {
        LambdaEsQueryWrapper<VolumeEs> volWrapper = new LambdaEsQueryWrapper<>();
        volWrapper.in(VolumeEs::getProjId, projIds);
        volumeEsSvc.remove(volWrapper);
      }
      if (BooleanUtil.isTrue(archiveTypeVO.getHaveBox())) {
        LambdaEsQueryWrapper<BoxEs> boxWrapper = new LambdaEsQueryWrapper<>();
        boxWrapper.in(BoxEs::getProjId, projIds);
        boxEsSvc.remove(boxWrapper);
      }
    }
    documentEsSvc.remove(wrapper);
  }
}
